package com.muchfish.reactor.api;

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

public class FluxSinkManualRequestExample {
    public static void main(String[] args) {
        Flux.create(fluxSink -> {
                    for (int i = 0; i < 100; i++) {
                        fluxSink.next(i);

                        if (i % 10 == 0 && fluxSink.requestedFromDownstream() == 0) {
                            // 当请求的元素达到 0 时，等待下游再次请求
                            while (fluxSink.requestedFromDownstream() == 0) {
                                // 等待下游请求
                                try {
                                    Thread.sleep(10);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    }
                    fluxSink.complete();
                })
                .subscribe(System.out::println);
    }
}
